package com.zhaosc.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Durations
/**
 * UpdateStateByKey的主要功能:
 * 1、Spark Streaming中为每一个Key维护一份state状态，state类型可以是任意类型的的， 可以是一个自定义的对象，那么更新函数也可以是自定义的。
 * 2、通过更新函数对该key的状态不断更新，对于每个新的batch而言，Spark Streaming会在使用updateStateByKey的时候为已经存在的key进行state的状态更新
 * 假设batch interval是5s
 * 第6s
 * 		hello 5
 * 		bjsxt 3
 * 第11s
 * 		hello 1
 * 		hello 1
 * 		hello 1
 * 		bjsxt 1
 * 		bjsxt 1
 * 		hello 【1,1,1】
 * 		bjsxt [1,1]
 *
 * 		hello 5+1+1+1=8
 * 		bjsxt 3+1+1=5
 * 。。。。。
 * 有何用？   统计广告点击流量，统计这一天的车流量，统计。。。。点击量
 */
object UpdateStateByKeyOperator {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("UpdateStateByKeyOperator").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(conf, Durations.seconds(5)) //5秒处理一次消息

    ssc.checkpoint("D:\\tmp\\updatestatebykeyoperator")
    val topic = Map("test" -> 1);
    // 这里的日志格式，就简化一下，就是date username的方式
    val kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181", "MyFirstConsumerGroup", topic, StorageLevel.MEMORY_AND_DISK);

    val result = kafkaStream.flatMap(v => {
      v._2.split(" ")
    }).map(v => {
      Tuple2(v, 1)
    }).updateStateByKey((vs: Seq[Int], o: Option[Int]) => {
      var count = o.getOrElse(0)
      for (v <- vs) {
        count = count + v
      }
      Option(count);
    })
    
    
    result.print();
    ssc.start();
    ssc.awaitTermination();
  }
}